package com.quanxiaoha.xiaohashu.data.align.job;

import cn.hutool.core.collection.CollUtil;
import com.quanxiaoha.xiaohashu.data.align.constant.RedisKeyConstants;
import com.quanxiaoha.xiaohashu.data.align.constant.TableConstants;
import com.quanxiaoha.xiaohashu.data.align.domain.mapper.DeleteMapper;
import com.quanxiaoha.xiaohashu.data.align.domain.mapper.SelectMapper;
import com.quanxiaoha.xiaohashu.data.align.domain.mapper.UpdateMapper;
import com.quanxiaoha.xiaohashu.data.align.rpc.SearchRpcService;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.List;

@Component
@Slf4j
public class NoteLikeCountShardingXxlJob {

    @Resource
    private SelectMapper selectMapper;
    @Resource
    private UpdateMapper updateMapper;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private DeleteMapper deleteMapper;
    @Resource
    private SearchRpcService searchRpcService;
    /**
     * 分片广播任务
     */
    @XxlJob("noteLikeCountShardingJobHandler")
    public void noteLikeCountShardingJobHandler() throws Exception {
        // 获取分片参数 -- 序号 和 总数
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();

        XxlJobHelper.log("=================> 开始定时分片广播任务：对当日发生变更的笔记点赞数进行对齐");
        XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);

        log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);

        // 表后缀
        String date = LocalDate.now().minusDays(1)
                .format(DateTimeFormatter.ofPattern("yyyyMMdd"));
        // 表名后缀
        String tableNameSuffix = TableConstants.buildTableNameSuffix(date,shardIndex);

        // 一批次 1000 条
        int batchSize = 1000;
        // 共对齐了多少条记录，默认为0
        int processedTotal = 0;

        for (;;) {
            // 1、分批次查询，如一批次查询1000条，直到查询到为空结束循环
            List<Long> noteIds = selectMapper.selectBatchFromDataAlignNoteLikeCountTempTable(tableNameSuffix, batchSize);
            // 记录为空终止循环
            if(CollUtil.isEmpty(noteIds)) break;
            // 遍历查询数据
            noteIds.forEach(noteId -> {
                // 2、对t_note_like 表执行 count(*) 操作，获取点赞总数
                int noteLikeTotal = selectMapper.selectCountFromNoteLikeTableByUserId(noteId);
                // 3: 更新 t_note_count 表, 更新对应 Redis 缓存
                int count = updateMapper.updateNoteLikeTotalByNoteId(noteId, noteLikeTotal);

                if (count > 0) {
                    String redisKey = RedisKeyConstants.buildCountNoteKey(noteId);
                    // 判断 Hash 是否存在
                    boolean hashKey = redisTemplate.hasKey(redisKey);
                    if (hashKey) {
                        // 更新 hash 中的 field 笔记点赞数
                        redisTemplate.opsForHash().put(redisKey, RedisKeyConstants.FIELD_LIKE_TOTAL, noteLikeTotal);
                    }
                }

                // 远程 RPC, 调用搜索服务，重新构建文档
                searchRpcService.rebuildNoteDocument(noteId);
            });
            // 批量删除当天记录
            deleteMapper.batchDeleteDataAlignNoteLikeCountTempTable(tableNameSuffix, noteIds);
            // 当前已处理记录数
            processedTotal += noteIds.size();
        }

        XxlJobHelper.log("=================> 结束定时分片广播任务：对当日发生变更的笔记点赞数进行对齐，共对齐记录数：{}", processedTotal);
    }
}
